package com.cnjson.sensor.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

//import com.cnjson.common.utils.IdGen;
import com.cnjson.sensor.config.Global;
import com.cnjson.sensor.db.entity.DataRecord;
import com.cnjson.sensor.db.entity.EquipmentState;
import com.cnjson.sensor.db.service.DataRecordService;
import com.cnjson.sensor.db.service.EquipmentStateService;
import com.cnjson.sensor.db.utils.SetIdUtils;
import com.cnjson.sensor.util.ByteUtils;
import com.cnjson.sensor.util.CRC16;
import com.cnjson.sensor.util.DataFormater;
import com.mysql.jdbc.StringUtils;
import com.sun.javafx.image.ByteToIntPixelConverter;

public class SocketServer {

	private static final Logger logger = LoggerFactory.getLogger(SocketServer.class);

	private static Map<String, SocketChannel> serverSocketMap = new HashMap<String, SocketChannel>();// 记录socket的键值对

	private static int status = 0;

	/**
	 * 创建非阻塞服务器绑定2020端口
	 */
	public SocketServer() {
		try {
			ss = ServerSocketChannel.open();
			Integer port = Global.getServerPort();
			ss.bind(new InetSocketAddress(port));
			ss.configureBlocking(false);
			selector = Selector.open();
			ss.register(selector, SelectionKey.OP_ACCEPT);

		} catch (Exception e) {
			logger.error(e.getMessage());
			close();
		}
	}

	private Map<Integer, Integer> rejectedThreadCountMap = new ConcurrentHashMap<>();

	private RejectedExecutionHandler rejectedExecutionHandler = new RejectedExecutionHandler() {
		@Override
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			try {
				TimeUnit.MILLISECONDS.sleep(100);
				int hashcode = r.hashCode();
				Integer count = rejectedThreadCountMap.get(hashcode);
				if (count == null) {
					count = 0;
					rejectedThreadCountMap.put(hashcode, count);
				} else {
					count++;
					rejectedThreadCountMap.put(hashcode, count);
				}
				if (count < 1) {
					executor.execute(r);
				} else {
					if (r instanceof WriteClientSocketHandler) {
						WriteClientSocketHandler realThread = (WriteClientSocketHandler) r;
						logger.info("服务系统繁忙,客户端WriteClientSocketHandler[" + realThread.client + "]请求被拒绝处理！");

						SelectionKey selectionKey = realThread.client.keyFor(selector);
						if (selectionKey != null) {
							selectionKey.cancel();
						}
						if (realThread.client != null) {
							try {
								realThread.client.close();
							} catch (IOException e) {
								e.printStackTrace();
							}
						}
						AtomicBoolean isWriting = isWritingMap.get(realThread.client);
						isWriting.set(false);
					} else if (r instanceof ReadClientSocketHandler) {
						ReadClientSocketHandler realThread = (ReadClientSocketHandler) r;
						logger.info("服务系统繁忙,客户端ReadClientSocketHandler[" + realThread.client + "]请求被拒绝处理！");
						SelectionKey selectionKey = realThread.client.keyFor(selector);
						if (selectionKey != null) {
							selectionKey.cancel();
						}
						if (realThread.client != null) {
							try {
								realThread.client.close();
							} catch (IOException e) {
								e.printStackTrace();
							}
						}
						AtomicBoolean isReading = isReadingMap.get(realThread.client);
						isReading.set(false);
					} else {
						logger.info("服务系统繁忙,系统线程[" + r.getClass().getName() + "]被拒绝处理！");
					}
				}
			} catch (InterruptedException e) {
				logger.info("RejectedExecutionHandler处理发生异常：" + e.getMessage());
			}
		}
	};

	private ThreadPoolExecutor threadPool = new ThreadPoolExecutor(30, 50, 300, TimeUnit.MILLISECONDS,
			new ArrayBlockingQueue<Runnable>(5), rejectedExecutionHandler);
	private Map<SocketChannel, AtomicBoolean> isReadingMap = new ConcurrentHashMap<SocketChannel, AtomicBoolean>();
	private static Map<SocketChannel, AtomicBoolean> isWritingMap = new ConcurrentHashMap<SocketChannel, AtomicBoolean>();
	private static Selector selector = null;
	private ServerSocketChannel ss = null;
	private volatile boolean isClose = false;

	public boolean isClose() {
		return isClose;
	}

	/**
	 * 关闭服务器
	 */
	private void close() {
		isClose = true;
		threadPool.shutdown();
		try {
			if (ss != null) {
				ss.close();
			}
			if (selector != null) {
				selector.close();
			}
		} catch (IOException e) {
			logger.error("服务器关闭发生异常：" + e.getMessage());
		}
	}

	/**
	 * 启动选择器监听客户端事件
	 */
	private void start() {
		threadPool.execute(new SelectorGuardHandler());
	}

	/**
	 * Nio 通道选择器
	 *
	 * @author cgli
	 */
	private class SelectorGuardHandler implements Runnable {

		@Override
		public void run() {

			while (!isClose) {
				try {
					if (selector.select(10) == 0) {
						continue;
					}
					Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
					while (iterator.hasNext()) {
						SelectionKey selectedKey = iterator.next();
						iterator.remove();
						try {
							if (selectedKey.isReadable()) {
								SocketChannel socketChannel = (SocketChannel) selectedKey.channel();
								AtomicBoolean isReading = isReadingMap.get(socketChannel);
								if (isReading == null) {
									isReading = new AtomicBoolean(false);
									isReadingMap.put(socketChannel, isReading);
								}
								while (isReading.get()) {
									Thread.sleep(5);
								}
								isReading.set(true);
								threadPool.execute(new ReadClientSocketHandler(socketChannel));

							} else if (selectedKey.isWritable()) {
								Object responseMessage = selectedKey.attachment();
								SocketChannel socketChannel = (SocketChannel) selectedKey.channel();
								selectedKey.interestOps(SelectionKey.OP_READ);
								System.out.println("---------------------->isWritable");
								logger.info("---------------------->isWritable");
								threadPool.execute(new WriteClientSocketHandler(socketChannel, responseMessage));

							} else if (selectedKey.isAcceptable()) {
								ServerSocketChannel ssc = (ServerSocketChannel) selectedKey.channel();
								SocketChannel clientSocket = ssc.accept();
								clientSocket.configureBlocking(false);
								logger.info("---------------------->isAcceptable");
								clientSocket.register(selector, SelectionKey.OP_READ);
							}
						} catch (CancelledKeyException e) {
							selectedKey.cancel();
							logger.error("服务器启动或运行发生异常：" + e);
						}
					}
				} catch (Exception e) {
					if (e instanceof NullPointerException) {
						e.printStackTrace();
						logger.error(e.getMessage());
						close();
					} else {
						logger.error("服务器启动或运行发生异常：" + e);
						close();
					}
					break;
				}
			}
		}
	}

	/**
	 * 响应数据给客户端线程
	 */
	private class WriteClientSocketHandler implements Runnable {
		private SocketChannel client;

		private Object responseMessage;

		private WriteClientSocketHandler(SocketChannel client, Object responseMessage) {
			this.client = client;
			this.responseMessage = responseMessage;
		}

		@Override
		public void run() {
			try {
				logger.info("------------------w1 准备写入数据-------------");
				byte[] responseByteData = null;
				String logResponseString = "";
				if (responseMessage instanceof byte[]) {
					responseByteData = (byte[]) responseMessage;
					logger.info("---------------w2 写字节数据-------------");
					logResponseString = DataFormater.bytesToHexString(responseByteData);
				} else if (responseMessage instanceof String) {
					logResponseString = (String) responseMessage;
					responseByteData = logResponseString.getBytes();
				} else if (responseMessage != null) {
					logger.info("不支持的数据类型" + responseMessage.getClass());
					return;
				}
				if (responseByteData == null || responseByteData.length == 0) {
					logger.info("服务器响应的写入数据为空");
					return;
				}
				client.write(ByteBuffer.wrap(responseByteData));

				logger.info("服务端响应客户端[" + client.getRemoteAddress() + "]写入数据 :[" + logResponseString + "]");
			} catch (Exception e) {
				try {
					logger.info("server响应客户端[" + client.getRemoteAddress() + "]数据 异常[" + e.getMessage() + "]");
					SelectionKey selectionKey = client.keyFor(selector);
					if (selectionKey != null) {
						selectionKey.cancel();
					}
					if (client != null) {
						client.close();
					}
				} catch (IOException e1) {
					e1.printStackTrace();
				}
			} finally {
				AtomicBoolean isWriting = isWritingMap.get(client);
				isWriting.set(false);
			}
		}
	}

	/**
	 * 读客户端发送数据线程
	 *
	 */
	private class ReadClientSocketHandler implements Runnable {

		private SocketChannel client;
		private ByteBuffer dataLen = ByteBuffer.allocate(4);//建了一个容量为4字节的ByteBuffer对象

		private ReadClientSocketHandler(SocketChannel client) {
			this.client = client;
		}

		@Override
		public void run() {
			ByteBuffer data = null;
			try {
				//2b 43 61 1 4f fc 0 0 2 0 3 12 c 14 10 37 12 0 0 0 12 0 0 2 58 0 0 2 58 1f 0 1 27 44 0 0 0 3 45 6c 72 67 69 d fc fc
				dataLen.clear();
				client.read(dataLen);
				if (dataLen.array().length > 0 && dataLen.hasArray()) {
					String slen = DataFormater.byteToHex(dataLen.get(3));//转成16进制
					int dataLength = Integer.parseInt(slen, 16);
					if (dataLength <= 0) {
						return;
					}

					// CRC校验数据需要带上 长度位,即 长度到负载的CRC校验
					ByteBuffer crcCheckData = ByteBuffer.allocate(dataLength - 1);
					crcCheckData.put(dataLen.get(3));
					data = ByteBuffer.allocate(dataLength + 2);
					while (data.hasRemaining()) {
						client.read(data);
					}
					// 数据长度值（帧长度） = [帧代号1]+[功能码2]+[地址段m]+[数据n]+[CRC校验2]
					int subLeng = data.array().length;
					StringBuilder sb = new StringBuilder();
					sb.append(slen);
					for (int i = 0; i < subLeng; i++) {
						sb.append(" ");
						byte b = data.get(i);
						sb.append(DataFormater.byteToHex(b));
						if (i < subLeng - 4) {//crc校验数据
							crcCheckData.put(b);
						}
					}
					logger.info(Thread.currentThread().getId() + " 接收数据 :[" + sb.toString() + "]");
					boolean isOK = checkCRCVal(crcCheckData, data);
					//测试数据
					if (!isOK) {
						logger.warn("CRC验证不通过!");
						return;
					}
					byte[] response = null;

					// 帧代号1
					int flag = Integer.parseInt(DataFormater.byteToHex(data.get(0)), 16);

					EquipmentStateService equipmentService = new EquipmentStateService();
					DataRecordService service = new DataRecordService();

					byte[] relDevice = getRelDevice(data); // 2位设备地址段
					String address = hexStringToString(relDevice);
					System.out.println("地址段：" + address);

					String remote = client.getRemoteAddress().toString();
//					address = "997";//测试数据
					String id = equipmentService.getByAddress(address);//根据equipment_id获取当前设备状态记录的id
					boolean isExistEquipment = equipmentService.isExistEquipment(address);//根据number字段查询设备表，确认设备存在
					if (isExistEquipment) {//设备是否存在 address为设备编号
						if (equipmentService.isExistEquipmentInStateTable(address)) {//根据equipment_id查询当前的设备状态是否存在
							EquipmentState eq = new EquipmentState(id);
							//
							eq.setEquipmentId(address);//设置设备的equipment_id
							eq.setRunningState("1");
							eq.setRemoteAddress(remote);
							equipmentService.updateState(eq);//根据equipment_id更新设备状态数据
						} else {
							Date date = new Date();
							EquipmentState e = new EquipmentState(new SetIdUtils().setId(), address, "1", date, date,
									date, date, "0", remote);
							equipmentService.addEquipmentState(e);
						}
					}
					id = equipmentService.getByAddress(address);

					if (serverSocketMap == null || (serverSocketMap != null && !serverSocketMap.containsKey(remote))) {
						serverSocketMap.put(remote, client);
					}

					// 设备登录
//					flag = 0x47;//测试数据
					if (flag == 0x41) {

						try {
							byte[] device = getRelDevice(data); // 设备地址段

							logger.info("---------0x41 设备登录注册-------------地址-----------------" + address);

							if (isExistEquipment) {
								logger.info("---------0x41 设备登录注册-------------");
							} else {
								logger.warn("---------0x41 设备未配置-------------");
							}
							response = buildRegisterPackage(data, isExistEquipment);
						} catch (Exception e) {
							logger.error("----------0x41-----------" + e.getMessage());
						}

						// 设备心跳
					} else if (flag == 0x43) {

						// 心跳类型 1：链路心跳，2：状态心跳
						/*
						 * response = new byte[subLeng + 4];
						 * System.arraycopy(dataLen.array(), 0, response, 0, 4);
						 * System.arraycopy(data.array(), 0, response, 4,
						 * subLeng); response[2] = (byte) 0xAF; response[6] =
						 * 0x00; byte[] crcTemp = new byte[subLeng - 3];
						 * System.arraycopy(response, 3, crcTemp, 0,
						 * response.length - 7); byte[] nR =
						 * CRC16.getCrcBytes(crcTemp); System.arraycopy(nR, 0,
						 * response, response.length - 4, 2);
						 */
						try {

							int type = Integer.parseInt(DataFormater.byteToHex(data.get(7)), 16);
							byte[] content = null;// 应答内容，不同类型不同内容
							byte[] device = getDevice(data); // 4位设备地址段
							// 链路心跳
							if (type == 1) {
								content = new byte[] { 0x01, 0x00 };
							} else if (type == 2) {
								content = new byte[4];
								content[0] = 0x02;
								content[1] = 0x00;
								content[2] = 0x00;
								content[3] = 0x00;
							} else {
								content = new byte[23];
								content[0] = 0x02;
								content[1] = 0x00;

								content[2] = 0x00;// 如果要同步时间，必须修改1，同时把正确时间下发。
								content[3] = 0x00;
								System.arraycopy(data.array(), 10, content, 4, 19);
							}
							response = buildResponsePackage((byte) 0x43, device, content);

							// update col_equipment_state
							// 判断记录是否存在用于insert/update

							updateEquipmentStateByHeart(data, service, equipmentService, id, address, remote);

							logger.info("----------0x43 接收设备心跳并返回:" + DataFormater.bytesToHexString(response) + "----");
						} catch (Exception e) {
							logger.error("----------0x43异常：-----------" + e);
						}
						// 客户端数据上报
					} else if (flag == 0x47) {
						byte[] device = getDevice(data); // 4位设备地址段

						String equipmentId = service.getEquipmentIdByAddress(address);
						if (StringUtils.isNullOrEmpty(equipmentId)) {
							System.out.println("---------0x47 设备未配置-------------");
							logger.warn("---------0x47 设备未配置-------------");
							return;
						}

						byte[] date = getData(data, 9, 6);// 日期时间段（年月日时分秒如：1608）
						Date sDate = DataFormater.bytesToDate(date);
						logger.info("-----------打印时间:" + sDate + "-----,设备地址---" + address);

						// SAVE TO DB.....
						try {
							// 通道数，一般固定为11个通道
							byte[] channelNum = getData(data, 15, 1);
							String number = byteToHexString(channelNum);
							if (number.length() < 2) {
								number = "0" + number;
							} else
								;
							// logger.info("-----------通道string:" + temp);
							int num = Integer.valueOf(number, 16);// ByteUtils.getInt(channelNum);

//							num = 9;//测试数据

							logger.info("-----------通道数:" + num);
							// 通道数据   通道的数量对应的是采集项的数量
							List<DataRecord> records = new ArrayList<>();
							List<Object> checkItemIds = service.getCheckItemIds(equipmentId, num);//获取采集项数据id串,允许实际的通道数量小于配置的

							//根据通道数量
							for (int j = 0; j < num; j++) {
								// 前三个设备状态信息 获取指定长度的数据
								byte[] d1 = getData(data, 16 + (j * 4), 4);//不同通道数据间隔

								String strres = byteToHexString(d1);
								Float f = 0f;
								try {
									//获取检测的数据
									f = Float.intBitsToFloat(Integer.valueOf(strres, 16));
								} catch (Exception e) {

								}

								/**
								 * 电压mv; 电导率为ms; 溶解氧为mg/l ;温度为℃ ; 其他没有单位
								 */
								// String unit = "";
								// if (j == 0) {
								// unit = "mv";
								// } else if (j == 2) {
								// unit = "ms";
								// } else if (j == 3 || j == 4) {
								// unit = "℃";
								// } else if (j == 5 || j == 6) {
								// unit = "mg/l";
								// }
								DataRecord record = new DataRecord();
								record.setChannelNO(j + 1);// 通道序号，并不是ID号
								// record.setRawData(f);// 原始值
								// record.setDataUnit(unit);
								if (checkItemIds.size() == num)
									record.setCheckItemId(checkItemIds.get(j).toString());
								else
									record.setCheckItemId("");
								record.setProcessedData(f);//
								record.setEquipmentId(equipmentId);// 是设备地址，需要转化为设备ID
								record.setAcquisitionTime(sDate);
								record.setReportingTime(sDate);
								records.add(record);
								logger.info("采集的数据：" + record.toString());
							}

							//上报成功标示：0xCB 0xCE
							byte[] content = new byte[] { (byte) 0xCB, (byte) 0xCE };
							response = buildResponsePackage((byte) 0x47, device, content);
							logger.info("-----------0x47 接收设备上报数据并返回----------");
							service.addBatch(records);
							logger.info("-----------数据保存至数据库----------");
						} catch (Exception e) {
							e.printStackTrace();
							logger.error("----------0x47-----------" + e.getMessage());
						}

					} else if (flag == 0x29) {

						logger.warn("-----------0x29 ----------");

					} else if (flag == 0x30) {// 索要数据
						logger.warn("-----------0x30 ----------");

					} else if (flag == 0x31) {// 状态数据上报
						logger.warn("-----------0x31 ----------");

					} else if (flag == 0x32) {// 警情上报
						logger.warn("-----------0x32 ----------");

					} else if (flag == 0x02) {// 开启设备（应答）
						logger.warn("-----------0x02 ----------" + sb.toString());
						final byte[] equipmentNo = this.getData(data, 3, 2);
						final byte[] channelNo = this.getData(data, 7, 1);
						final byte[] state = this.getData(data, 8, 1);
						final EquipmentStateService equipmentStateService = EquipmentStateService.getInstance();
						//更新控制设备状态日志
						equipmentStateService.updateCtrlEquipmentState(String.valueOf(Integer.parseInt(SocketServer.byteToHexString(state), 16)), String.valueOf(Integer.parseInt(SocketServer.byteToHexString(equipmentNo), 16)), String.valueOf(Integer.parseInt(SocketServer.byteToHexString(channelNo), 16) + 1));
						final String equipment = SocketServer.hexStringToString(equipmentNo);
						final String channel = String.valueOf(Integer.parseInt(SocketServer.hexStringToString(channelNo)) + 1);
						//控制成功，修改日志
						equipmentService.updateCtrlEquipmentLog(equipment, channel, "1", "0");
						//修改控制设备状态
						equipmentService.updateCtrlEquipment(equipment + "-" + channel,"1");
					} else if (flag == 0x82) {// 关闭设备（应答）
						try {
							SocketServer.logger.warn("-----------0x82 ----------" + sb.toString());
							final byte[] equipmentNo = this.getData(data, 3, 2);
							final byte[] channelNo = this.getData(data, 7, 1);
							final byte[] state = this.getData(data, 8, 1);
							final EquipmentStateService eq2 = EquipmentStateService.getInstance();
							eq2.updateCtrlEquipmentState(String.valueOf(Integer.parseInt(SocketServer.byteToHexString(state), 16)), String.valueOf(Integer.parseInt(SocketServer.byteToHexString(equipmentNo), 16)), String.valueOf(Integer.parseInt(SocketServer.byteToHexString(channelNo), 16) + 1));
							final String equipment = SocketServer.hexStringToString(equipmentNo);
							final String channel = String.valueOf(Integer.parseInt(SocketServer.hexStringToString(channelNo)) + 1);
							//控制成功，修改日志
							equipmentService.updateCtrlEquipmentLog(equipment, channel, "1", "0");
							//修改控制设备状态
							equipmentService.updateCtrlEquipment(equipment + "-" + channel,"0");
						}
						catch (Exception e2) {
							SocketServer.logger.error("************" + e2.getMessage(), (Throwable)e2);
						}
					}
					if (response != null) {
						AtomicBoolean isWriting = isWritingMap.get(client);
						if (isWriting == null) {
							isWriting = new AtomicBoolean(false);
							isWritingMap.put(client, isWriting);
						}
						while (isWriting.get()) {
							Thread.sleep(5);
						}
						isWriting.set(true);
						client.register(selector, SelectionKey.OP_WRITE, response);
					} else {
						logger.warn("-----------服务端没有可应答的数据！----------");
					}
				}
			} catch (Exception e) {
				try {
					SelectionKey selectionKey = client.keyFor(selector);
					if (selectionKey != null) {
						selectionKey.cancel();
					}

					String address = client.getRemoteAddress().toString();
					EquipmentStateService eq = new EquipmentStateService();
					eq.updateRunningState(false, address);

					serverSocketMap.remove(address);

					logger.warn("客户端[" + client + "]关闭了连接,原因[" + e + "]");
					if (client != null) {
						client.close();
					}
				} catch (Exception e1) {
					logger.error("客户端[" + client + "]关闭异常");// + e1.getMessage()
				}
			} finally {
				AtomicBoolean isReading = isReadingMap.get(client);
				isReading.set(false);
			}
		}


		/**
		 * 对比数据CRC结果
		 *
		 * @param crcCheckData
		 *            待校验的数据
		 * @param data
		 *            原来去包头（不包含前四位:起始位和方向位）的数据包。
		 * @return 一致则true, 否则false
		 */

		private boolean checkCRCVal(final ByteBuffer crcCheckData, final ByteBuffer data) {
			int total = data.array().length;
			String t1 = DataFormater.byteToHex(data.get(total - 4));
			String t2 = DataFormater.byteToHex(data.get(total - 3));
			if (t1.length() < 2) {
				t1 = "0" + t1;
			}
			if (t2.length() < 2) {
				t2 = "0" + t2;
			}
			String originCrc = t1 + t2;
			//注释掉原来的用replace的方法
			String crc = CRC16.getCRC(crcCheckData.array());
//			if (crc.length() == 4) {
//				
//	        } else if (crc.length() == 3) {
//	        	crc = "0" + crc;
//	        } else if (crc.length() == 2) {
//	        	crc = crc+"00";
//	        }
			logger.info("原数据CRC值=" + originCrc + ",计算CRC结果 :[" + crc + "]");
			if (crc.equalsIgnoreCase(originCrc)) {
				return true;
			}
			return false;
		}

		/**
		 * 获取设备在址
		 *
		 * @param data
		 *            去掉包头（不包含前四位）的数据包
		 * @return 返回设备地址4位byte
		 */
		private byte[] getDevice(ByteBuffer data) {
			byte[] device = new byte[4];
			System.arraycopy(data.array(), 3, device, 0, 4);
			return device;
		}

		/**
		 * 获取设备在址
		 *
		 * @param data
		 *            去掉包头（不包含前四位）的数据包
		 * @return 返回设备地址2位byte
		 */
		private byte[] getRelDevice(ByteBuffer data) {
			byte[] device = new byte[2];
			System.arraycopy(data.array(), 3, device, 0, 2);
			return device;
		}

		/**
		 * 获取一个数据指定长度的内容段
		 *
		 * @param data
		 *            原数据
		 * @param posIndex
		 *            起如位置
		 * @param len
		 *            截取的长度
		 * @return 定长度的内容段 byte[]
		 */
		private byte[] getData(ByteBuffer data, int posIndex, int len) {
			byte[] b = new byte[len];
			System.arraycopy(data.array(), posIndex, b, 0, len);
			return b;
		}

		/**
		 * 构建应答设备登录（0x41）的数据包
		 * 注册成功：0x01 错误标志：0xCF 0xCF 0x01：未配置 0x02：数据内容错误
		 * @param data
		 *            去掉包头（不包含前四位）的数据包
		 * @return 完整的数据
		 */
		private byte[] buildRegisterPackage(ByteBuffer data, boolean exist) {
			byte[] device = getDevice(data);
			byte[] content;
			if (exist) {
				content = new byte[] { 0x01 };
			} else {
				content = new byte[] { (byte) 0xCF, (byte) 0xCF, 0x01 };
			}
			return buildResponsePackage((byte) 0x41, device, content);
		}

		/**
		 * 构建应答数据包。
		 *
		 * @param direction
		 *            标识位，即功能代码
		 * @param device
		 *            设备地址字节数组
		 * @param content
		 *            要应答数据内容
		 * @return
		 */
		private byte[] buildResponsePackage(byte direction, byte[] device, byte[] content) {
			// 响应数据包总长度=[帧头2] +[方向1]+[帧长度位1]＋[帧代号1]+[功能码2]+ [地址段4] + [数据n]
			// +[CRC校验2]+[包尾2]
			final int dlen = device.length;
			final int len = 7 + dlen + content.length + 2 + 2;
			byte[] response = new byte[len];
			response[0] = (byte) 0x88;
			response[1] = (byte) 0xFB;
			response[2] = (byte) 0xAF;
			// 长度值 = [帧代号1]+[功能码2] + [地址段m] + [数据n] +[CRC校验2]
			response[3] = (byte) (len - 6);// 数据包有效长度值，从长度位到负载的
			response[4] = direction;
			response[5] = (byte) 0x61;
			response[6] = (byte) 0x00;
			System.arraycopy(device, 0, response, 7, dlen);// 设备在址
			System.arraycopy(content, 0, response, 7 + dlen, content.length);// 应答内容
			byte[] resCrc = new byte[len - 7];
			System.arraycopy(response, 3, resCrc, 0, len - 7);
			byte[] nR = CRC16.getCrcBytes(resCrc);
			System.arraycopy(nR, 0, response, len - 4, 2);
			// 包尾
			response[len - 2] = (byte) 0xFC;
			response[len - 1] = (byte) 0xFC;
			return response;
		}

		/**
		 * 根据心跳更新设备状态
		 * 
		 * @param data
		 * @param service
		 * @param equipmentService
		 * @param id
		 * @param address
		 * @param remote
		 */
		private void updateEquipmentStateByHeart(ByteBuffer data, DataRecordService service,
				EquipmentStateService equipmentService, String id, String address, String remote) {
			byte[] heart = getData(data, 7, 1);
			byte[] link = getData(data, 9, 1);
			byte[] time = getData(data, 10, 6);
			byte[] reportCycle = getData(data, 16, 4);
			byte[] acquisionCycle = getData(data, 20, 4);
			byte[] storageCycle = getData(data, 24, 4);//存储周期
			byte[] signal = getData(data, 28, 1);
			byte[] storedData = getData(data, 29, 4);//存储数据
			byte[] unreportedData = getData(data, 33, 4);
			byte[] equipmentEnergy = getData(data, 37, 4);

			Date sDate = DataFormater.bytesToDate(time);
			String ee = byteToHexString(equipmentEnergy);

			int sd = Integer.parseInt(hexStringToString(storedData));
			int ud = Integer.parseInt(hexStringToString(unreportedData));

			Float energy = Float.intBitsToFloat(Integer.parseInt(ee, 16));

			String sig = hexStringToString(signal);
			String heartbean = hexStringToString(heart);
			String linkType = hexStringToString(link);
			String rc = hexStringToString(reportCycle);
			String ac = hexStringToString(acquisionCycle);
			String sc = hexStringToString(storageCycle);

			Date date = new Date();
			// EquipmentStateService service = new EquipmentStateService();
			String equipmentId = service.getEquipmentIdByAddress(address);
			if (StringUtils.isNullOrEmpty(equipmentId)) {
				logger.warn("---------0x43 设备未配置-------------");
				return;
			}

			if (!StringUtils.isNullOrEmpty(id)) {
				// update
				EquipmentState e = new EquipmentState(id);
				e.setRunningState("1");
				e.setCurTime(date);
				e.setCommunicationTime(sDate);
				e.setSignalIntensity(sig);
				e.setHeartbeatType(heartbean);
				e.setLinkType(linkType);
				e.setReportingCycle(rc);
				e.setAcquisionCycle(ac);
				e.setStorageCycle(sc);
				e.setStoredData(sd);
				e.setUnreportedData(ud);
				e.setEquipmentEnergy(energy);
				e.setUpdateBy("");
				e.setUpdateDate(date);
				e.setDelFlag("0");
				e.setRemoteAddress(remote);
				equipmentService.updateEquipmentState(e);
			} else {
				// insert
				EquipmentState e = new EquipmentState();
				e.setId(new SetIdUtils().setId());
				e.setEquipmentId(address);
				e.setRunningState("1");
				e.setCurTime(date);
				e.setCommunicationTime(sDate);
				e.setSignalIntensity(sig);
				e.setHeartbeatType(heartbean);
				e.setLinkType(linkType);
				e.setReportingCycle(rc);
				e.setAcquisionCycle(ac);
				e.setStorageCycle(sc);
				e.setStoredData(sd);
				e.setUnreportedData(ud);
				e.setEquipmentEnergy(energy);
				e.setCreateBy("");
				e.setCreateDate(date);
				e.setUpdateDate(date);
				e.setDelFlag("0");
				e.setRemoteAddress(remote);
				equipmentService.addEquipmentState(e);
			}
		}
	}  

	/**
	 * @param direction
	 *            标识位，即功能代码
	 * @param device
	 *            设备地址字节数组
	 * @param content
	 *            要应答数据内容
	 * @return
	 */
	private static byte[] buildRequestPackage(byte direction, byte[] device, byte[] content) {

		// 请求数据包总长度=[帧头2] +[方向1]+[帧长度位1]＋[帧代号1]+[功能码2]+ [地址段4] + [数据n]
		// +[CRC校验2]+[包尾2]
		final int dlen = device.length;
		final int len = 7 + dlen + content.length + 2 + 2;

		byte[] request = new byte[len];
		request[0] = (byte) 0x88;
		request[1] = (byte) 0xFB;
		request[2] = (byte) 0xFA;

		// 长度值 = [帧代号1]+[功能码2] + [地址段m] + [数据n] +[CRC校验2]
		request[3] = (byte) (len - 6);// 数据包有效长度值，从长度位到负载的
		request[4] = direction;
		request[5] = (byte) 0x41;
		request[6] = (byte) 0x01;
		System.arraycopy(device, 0, request, 7, dlen);// 设备地址
		System.arraycopy(content, 0, request, 7 + dlen, content.length);// 应答内容
		byte[] resCrc = new byte[len - 7];
		System.arraycopy(request, 3, resCrc, 0, len - 7);
		byte[] nR = CRC16.getCrcBytes(resCrc);
		System.arraycopy(nR, 0, request, len - 4, 2);

		// 包尾
		request[len - 2] = (byte) 0xFC;
		request[len - 1] = (byte) 0xFC;
		return request;
	}

	public static String byteToHexString(byte[] a) {
		StringBuilder bb = new StringBuilder();
		for (byte b : a) {
			String no = ByteUtils.byteToHex(b);
			if (no.length() < 2) {
				no = "0" + no;
			} else
				;
			bb.append(no);
		}
		return bb.toString();
	}

	public static String hexStringToString(byte[] b) {
		String str = byteToHexString(b);
		String sig = String.valueOf(Integer.parseInt(str, 16));
		return sig;
	}

	public static void main(String[] args) {
		logger.info("--------------->Enter to main method");
		SocketServer server = new SocketServer();
		server.start();

		logger.info("--------------->Server already started!");

		//定时任务
		while (true) {
			final SocketServer socketServer = server;
			socketServer.getClass();
			final Job job = socketServer.new Job();
			job.timeJob();
			try {
				Thread.sleep(1000L);
			} catch (Exception e) {
				SocketServer.logger.error("sleep == " + e);
			}
		}

		 /*TimerTask task = new TimerTask() {
			 @Override
			 public void run() {
				 // task to run goes here
				 timeJob();
			 }
		 };
		 Timer timer = new Timer();
		 long delay = 120000;
		 long intevalPeriod = 1 * 60000;
		 timer.scheduleAtFixedRate(task, delay, intevalPeriod);

//		 schedules the task to be run in an interval
		 SocketChannel socketChannel = (SocketChannel) selectedKey.channel();

		 int num = Integer.valueOf("1d", 16);

		 Float f = Float.intBitsToFloat(Integer.parseInt("4148f5c",16));//
		 byte[] d1=ByteUtils.hexStringToBytes("4148f5c");//410f78d5 41b75c29
//		 41209b23 c1546c7f

		 Float f = ByteUtils.bytesToFloat(d1);

//		 String s="410f78d5";
//		 Float f = Float.intBitsToFloat(Integer.valueOf(s.trim(), 16));
		//
		 System.out.print("--------------------="+num);
		 System.out.print(String.valueOf(Integer.valueOf("1d", 16)));
		 EquipmentStateService equipmentService = new EquipmentStateService();
		 boolean b = equipmentService.isExistEquipment("8106");
		 System.out.println(b);
		 DataRecordService service = new DataRecordService();
		 List<Object> checkItemIds =
		 service.getCheckItemIds("5ebb8a8608ca4c33a6ed00afaa4e99d6", 4);
		 System.out.println(checkItemIds.get(1).toString());*/
	}

	public static int byte2Int(byte[] b) {
		int intValue = 0;
		for (int i = 0; i < b.length; i++) {
			intValue += (b[i] & 0xFF) << (8 * (3 - i));
		}
		return intValue;
	}

	static /* synthetic */ void access$10(final int status) {
		SocketServer.status = status;
	}

	public class Job {

		public void timeJob() {
			byte[] request = null;
			final EquipmentStateService equipmentService = EquipmentStateService.getInstance();
			//获取为被“消费”的控制日志 0 -- 没有被消费 1 -- 已经被消费
			final Map<String, String> result = equipmentService.getMessageQueue("0");
			if (result == null) {
				System.out.println(new Date() + " ***** " + "队列为空");
				return;
			}
			final String logId = result.get("id");
			final String remarks = result.get("remarks");//控制通道编号/控制设备编号
			final String operateType = result.get("operateType");
			SocketServer.logger.warn("******" + remarks + " " + operateType + " " + logId);
			final String[] splits = remarks.split("-");//8106-KZD-2 8106-2
			final String equipmentNo = splits[0];
			final String channelNo = splits[1];
			final int equipmentNum = Integer.valueOf(equipmentNo);
			final int channelNum = Integer.valueOf(channelNo) - 1;
			final String equipmentHex = Integer.toHexString(equipmentNum);
			final String channelHex = Integer.toHexString(channelNum);
			final String high = equipmentHex.substring(0, 2);
			final String low = equipmentHex.substring(2, 4);
			final byte hb = (byte)Integer.parseInt(high, 16);
			final byte lb = (byte)Integer.parseInt(low, 16);
			equipmentService.updateCtrlEquipmentLogById(logId, remarks, "0");//控制状态初始为0
			try {
				SocketServer.logger.warn("________________控制开始————————————————————————" + equipmentNo);
				final byte[] content = { 0 };
				final byte[] device = { hb, lb, 0, 0 };
				request = buildRequestPackage((byte)Integer.parseInt(operateType.substring(2, 4), 16), device, content);
				if (operateType.equals("0x81") || operateType.equals("0x02")) {//正向脉冲或者高电平
					SocketServer.access$10(1);
				} else {
					SocketServer.access$10(0);
				}
				SocketChannel client = null;
				if (SocketServer.serverSocketMap != null) {
					client = SocketServer.serverSocketMap.get(equipmentService.getRemote(equipmentNo));
				}
				if (client != null) {
					AtomicBoolean isWriting = SocketServer.isWritingMap.get(client);
					if (isWriting == null) {
						isWriting = new AtomicBoolean(false);
						SocketServer.isWritingMap.put(client, isWriting);
					}
					if (isWriting != null && isWriting.get()) {
						Thread.sleep(5L);
						return;
					}
					if (isWriting != null) {
						isWriting.set(true);
					}
					client.register(SocketServer.selector, 4, request);
				}
				SocketServer.logger.warn("________________控制结束————————————————————————" + equipmentNo);
			}
			catch (Exception e) {
				SocketServer.logger.error("________________异常————————————————————————" + e);
			}
		}
	}



	/**
	 * 定时任务
	 */
	public static void timeJob() {
		byte[] request = null;
		// 控制设备
		try {
			byte[] relDevice = { 0x1f, (byte) 0xaa }; // 2位设备地址段
			String address = hexStringToString(relDevice);// String.valueOf(Integer.parseInt(strDevice,
			if (address.equals("8106")) {

				logger.warn("________________控制开始————————————————————————" + address);

				byte[] content2 = null;// 应答内容，不同类型不同内容
				byte[] device2 = { 0x1f, (byte) 0xaa, 0x00, 0x00 }; // 4位设备地址段

				content2 = new byte[1];
				content2[0] = 0x00; // 通道号 - 1

				request = buildRequestPackage(status == 0 ? (byte) 0x02 : (byte) 0x82, device2, content2);
				status = status == 0 ? 1 : 0;
				logger.warn("________________数据包:———————————————————————— " + DataFormater.bytesToHexString(request));

				EquipmentStateService equipmentService = new EquipmentStateService();
				SocketChannel client = null;
				if (serverSocketMap != null) {
					client = serverSocketMap.get(equipmentService.getRemote("8106"));
					logger.warn("==================serverSocketMap != null====================");
				}
				if (client != null) {
					AtomicBoolean isWriting = isWritingMap.get(client);
					if (isWriting == null) {
						isWriting = new AtomicBoolean(false);
						isWritingMap.put(client, isWriting);
					}
					while (isWriting.get()) {
						Thread.sleep(5);
					}
					isWriting.set(true);
					client.register(selector, SelectionKey.OP_WRITE, request);
				}
			}

		} catch (Exception e) {
			logger.warn("________________异常：————————————————————————" + e);
		}
	}



}
